#pragma once

#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"
#include "../mqcommon/mq_threadpool.hpp"
#include "google/protobuf/map.h"
#include "mq_virtualhost.hpp"
#include "mq_consumer.hpp"
#include "mq_route.hpp"


namespace bitmq
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr = std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr = std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr = std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr = std::shared_ptr<basicCancelRequest>;

    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;

        Channel(const std::string& id,
            const VirtualHost::ptr& host, 
            const ConsumerManager::ptr& cmp,
            const ProtobufCodecPtr& codec,
            const muduo::net::TcpConnectionPtr& conn,
            const threadpool::ptr& pool)
                : _cid(id)
                , _host(host)
                , _conn(conn)
                , _codec(codec)
                , _cmp(cmp)
                , _pool(pool)
        {
            DLOG("new Channel: %s", id.c_str());
        }

        ~Channel()
        {
            //删除信道对应的消费者
            if (_consumer.get() != nullptr)
            {
                _cmp->remove(_consumer->tag, _consumer->qname);
                DLOG("remove Channel: %s", _cid.c_str());
            }
        }

        //1）声明交换机
        void declareExchange(const declareExchangeRequestPtr& req)
        {
            bool ret = _host->declareExchange(req->exchange_name(), req->exchange_type(),
                req->durable(), req->auto_delete(), req->args());
            return basicResponse(ret, req->rid(), req->cid());
        }

        //2）删除交换机
        void deleteExchange(const deleteExchangeRequestPtr& req)
        {
            _host->deleteExchange(req->exchange_name());    
            return basicResponse(true, req->rid(), req->cid());
        }

        //3）声明队列
        void declareQueue(const declareQueueRequestPtr& req)
        {
            bool ret = _host->declareQueue(req->queue_name(), req->durable(), req->exclusive(),
                req->auto_delete(), req->args());
            if (ret == false)
                return;
            //初始化队列的消费者管理句柄
            _cmp->initQueueConsumer(req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        //4）删除队列
        void deleteQueue(const deleteQueueRequestPtr& req)
        {
            _host->deleteQueue(req->queue_name());
            _cmp->destroyQueueConsumer(req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        //5）队列的绑定
        void queueBind(const queueBindRequestPtr& req)
        {
            bool ret = _host->bind(req->exchange_name(), req->queue_name(),
                req->binding_key());
            if (ret == true)
            {
                DLOG("绑定队列成功");
            }
            else 
            {
                DLOG("绑定队列失败");
            }
            return basicResponse(ret, req->rid(), req->cid());
        }

        //6）队列的解绑定
        void queueUnBind(const queueUnBindRequestPtr& req)
        {
            _host->unBind(req->exchange_name(), req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

        //7）发布消息
        void basicPublish(const basicPublishRequestPtr req)
        {
            //1.判断交换机是否存在
            auto ep = _host->selectExchange(req->exchange_name());
            if (ep.get() == nullptr)
                return basicResponse(true, req->rid(), req->cid());
            //2.进行交换路由，判断消息可以发布到交换机哪个队列中
            MsgQueueBindingMap mqbm = _host->exchangeBindings(req->exchange_name());
            BasicProperties* bp = nullptr;
            std::string routing_key;
            //判断properties是否为空
            if (req->has_properties() != false)
            {
                routing_key = req->properties().routing_key();
                bp = req->mutable_properties();
            }
            for (auto& binding : mqbm)
            {
                if (Router::route(ep->type, routing_key, binding.second->binding_key))
                {
                    //消息匹配成功，则将该消息加入队列
                    _host->basicPublish(binding.first, bp, req->body());
                    //向线程池中添加一个消息消费任务（向指定队列的订阅者去推送消息————由线程池完成
                    auto task = std::bind(&Channel::publish, this, binding.first);
                    _pool->push(task);
                }
            }
            return basicResponse(true, req->rid(), req->cid());
        }

        //8）确认消息
        void basicAck(const basicAckRequestPtr& req)
        {
            _host->basicAck(req->queue_name(), req->message_id());
            return basicResponse(true, req->rid(), req->cid());
        }

        //9）订阅队列
        void basicConsumer(const basicConsumeRequestPtr& req)
        {
            //1.判断队列是否存在
            bool ret = _host->existsQueue(req->queue_name());
            if (ret == false)
            {
                DLOG("订阅队列失败，%s队列不存在", req->queue_name().c_str());
                return;
            }
            //2.创建队列消费者
            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1, 
                std::placeholders::_2, std::placeholders::_3);
            //在创建了消费者之后，当前的channel角色就是一个消费者
            //该消费者需要保存起来，以便在取消订阅 / 删除信道后删除对应的消费者（以防资源泄露）
            _consumer = _cmp->create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
            return basicResponse(true, req->rid(), req->cid());
        }

        //10）取消订阅
        void basicCancel(const basicCancelRequestPtr& req)
        {
            _cmp->remove(req->consumer_tag(), req->queue_name());
            return basicResponse(true, req->rid(), req->cid());
        }

    private:
        void callback(const std::string& tag,  const BasicProperties* bp, const std::string& body)
        {
            //针对参数组织处推送消息请求，将消息推送给channel对应的客户端
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_body(body);
            resp.set_consumer_tag(tag);
            if (bp != nullptr)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn, resp);
        }

        void publish(const std::string& qname)
        {
            //指定队列消费消息
            //1.从队列中获取一条消息
            MessagePtr mp = _host->basicConsume(qname);
            if (mp.get() == nullptr)
            {
                DLOG("执行消费任务时，%s队列没有消息", qname.c_str());
                return;
            }
            //2.从队列订阅者中取出一个订阅者
            Consumer::ptr cmp = _cmp->choose(qname);
            if (mp.get() == nullptr)
            {
                DLOG("执行消费任务时，%s队列没有订阅者", qname.c_str());
                return;
            }
            //3.调用订阅者对应的消息处理函数，实现消息的推送
            cmp->callback(cmp->tag, mp->mutable_payload()->mutable_properties(), mp->payload().body());
            //4.判断如果订阅者是自动确认——不需要等待确认，直接删除消息，否则需要收到确认后再删除
            if (cmp->auto_ack == true)
            {
                _host->basicAck(qname, cmp->tag);
            }
        }

        void basicResponse(bool ok, const std::string& rid, const std::string& cid)
        {
            basicCommonResponse resp;
            resp.set_rid(rid);
            resp.set_cid(cid);
            resp.set_is_ok(ok);
            _codec->send(_conn, resp);
        }

    private:
        std::string _cid;           // 1）信道 ID：信道的唯一标识
        Consumer::ptr _consumer;    // 2）信道关联的消费者
        muduo::net::TcpConnectionPtr _conn;  // 3）信道关联的连接：用于向客户端发送数据（响应，推送的消息）
        ProtobufCodecPtr _codec;      // 4）protobuf 协议处理句柄：网络通信前的协议处理
        ConsumerManager::ptr _cmp;  // 5）消费者管理句柄：信道关闭/取消订阅的时候，通过句柄删除订阅者信息
        VirtualHost::ptr _host;     // 6）虚拟机句柄：交换机/队列/绑定/消息数据管理
        threadpool::ptr _pool;      // 7）线程池
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;
        ChannelManager()
        {}

        bool openChannel(const std::string& cid,
            const VirtualHost::ptr& host, 
            const ConsumerManager::ptr& cmp,
            const ProtobufCodecPtr& codec,
            const muduo::net::TcpConnectionPtr& conn,
            const threadpool::ptr& pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(cid);
            if (it != _channels.end())
            {
                DLOG("信道：%s 已经存在!", cid.c_str());
                return false;
            }
            DLOG("创建信道成功: %s", cid.c_str());
            auto channel = std::make_shared<Channel>(cid, host, cmp, codec, conn, pool);
            _channels.insert(std::make_pair(cid, channel));
            return true;
        }

        void closeChannel(const std::string& id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _channels.erase(id);
        }

        Channel::ptr getChannel(const std::string& id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it == _channels.end())
            {
                return {};
            }
            return it->second;
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::ptr> _channels;
    };
}